package project.p1_car.util

import org.apache.commons.math3.random.{GaussianRandomGenerator, JDKRandomGenerator}
import org.apache.flink.table.expressions.LocalTime
import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord, RecordMetadata}

import java.text.SimpleDateFormat
import java.util.{Properties, Random}

object mkDataToDisk {


  val TOPIC = "flink_traffic_log"

  def main(args: Array[String]): Unit = {

    genCarData("2021-09-26", 1000)

  }


  /**
   *
   * @param day      eg. 2020-06-21
   * @param dataCount eg. 100
   * @return
   */
  def genCarData(day: String, dataCount: Int): Unit = {


    val prop = new Properties()

    prop.setProperty("bootstrap.servers", "127.0.0.1:9092")
    prop.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    prop.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    val producer: KafkaProducer[String, String] = new KafkaProducer(prop)


    val locations: Array[String] = Array("湘", "京", "京", "京", "京", "京", "京", "鲁", "皖", "鄂")

    val monitors: Array[String] = Array("0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010")

    //初始化高斯分布的对象
    val generator: JDKRandomGenerator = new JDKRandomGenerator
    generator.setSeed(10)
    val rg: GaussianRandomGenerator = new GaussianRandomGenerator(generator)

    val r: Random = new Random

    for (i <- 0 until dataCount) { //得到随机的车牌
      val car: String = locations(r.nextInt(locations.length)) + (65 + r.nextInt(26)).toChar + "%05d".format(r.nextInt(10000))

      //随机的小时
      var hour: String = "%02d".format(r.nextInt(24))

      //一天内，在一个城市里面，一辆车大概率经过30左右的卡口
      val v: Double = rg.nextNormalizedDouble //标准高斯分布的随机数据,大概率在-1 到 1

      val count: Int = (Math.abs(30 + 30 * v) + 1).toInt //一辆车一天内经过的卡口数量

      for (j <- 0 until count) { //如果count超过30，为了数据更加真实，时间要1小时

        if (j % 30 == 0) {
          var newHour: Int = hour.toInt + 1
          if (newHour == 24) newHour = 0
          hour = "%02d".format(newHour)
        }

        //经过卡口的时间
        val time: String = day + " " + hour + ":" + "%02d".format(r.nextInt(60)) + ":" + "%02d".format(r.nextInt(60))
//        val actionTime: Long = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(time).getTime

        val actionTime = System.currentTimeMillis();
        //卡口ID
//        val monitorId: String = "%04d".format(r.nextInt(10000))

        val monitorId = monitors(r.nextInt(monitors.length))

        //随机道路
        val roadId: String = "%02d".format(r.nextInt(100))
        val areaId: String = "%02d".format(r.nextInt(100))
        val cameraId: String = "%05d".format(r.nextInt(100000))

        //随机速度
        val v2: Double = rg.nextNormalizedDouble
        val speed: String = "%.1f".format(Math.abs(40 + (40 * v2))) //均值40，标准差是30


        val data: String = actionTime + "," + monitorId + "," + cameraId + "," + car + "," + speed + "," + roadId + "," + areaId



        producer.send(new ProducerRecord[String, String](TOPIC, null, data), new Callback {
          override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
            if (exception != null) {
              exception.printStackTrace()
            }
//            println("offsets:" + metadata.offset())
          }
        })

        Thread.sleep(200)

      }
    }

  }

}